/*
 * @Author: error: error: git config user.name & please set dead value or install git && error: git config user.email & please set dead value or install git & please set dead value or install git
 * @Date: 2023-11-24 10:27:58
 * @LastEditors: error: error: git config user.name & please set dead value or install git && error: git config user.email & please set dead value or install git & please set dead value or install git
 * @LastEditTime: 2023-11-27 20:16:05
 * @FilePath: /logs/looper.hpp
 * @Description: 这是默认设置,请设置`customMade`, 打开koroFileHeader查看配置 进行设置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
 */
/*实现异步工作器*/
#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__

#include "buffer.hpp"
#include <thread>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <functional>
#include <memory>

namespace zbtlog {
    enum class AsyncType {
        ASYNC_SAFE,//安全状态，表示缓冲区满了则返回阻塞，避免资源耗尽的风险
        ASYNC_UNSAFE//不考虑资源耗尽的问题，无限扩容，通常使用与压力测试
    };
    class AsyncLooper {
        public:
            using Functor = std::function<void(Buffer &buffer)>;
            using ptr = std::shared_ptr<AsyncLooper>;
            AsyncLooper(const Functor &cb, AsyncType loop_type = AsyncType::ASYNC_SAFE):
                _stop(false),
                _looper_type(loop_type),
                _thread(std::thread(& AsyncLooper::threadEntry, this)),
                _callBack(cb) {}
            
            void stop() {
                _stop = true;//将退出标志设置为true
                _cond_con.notify_all();//唤醒所有的工作线程
                _thread.join();//等待工作线程的退出
            }
            ~AsyncLooper(){ stop(); }

            void push(const char *data, size_t len) {
                //1.无限扩容-非安全； 2.固定大小--生产缓冲区中数据满了就阻塞
                std::unique_lock<std::mutex> lock(_mutex);
                //条件变量控制，若缓冲区剩余空间大小大于数据长度，则可以添加数据
                if(_looper_type == AsyncType::ASYNC_SAFE)
                    _cond_con.wait(lock, [&](){ return _pro_buf.writeAbleSize() >= len;  });
                //能够走下来代表满足了条件，可以向缓冲区添加数据
                _pro_buf.push(data, len);
                //唤醒消费者对缓冲区中的数据进行处理
                _cond_con.notify_one();
            }
        private:
            //线程入口函数--对消费缓冲区中的数据进行处理，处理完毕后，初始化缓冲区，交换缓冲区
            void threadEntry() {
                while(1) {
                    //为互斥锁设置一个生命周期，当缓冲区交换完毕后就解锁 (并不对数据处理过程加锁保护)
                    {
                        //1.判断生产缓冲区中有没有数据，有则交换，无则阻塞
                        std::unique_lock<std::mutex> lock(_mutex);
                        if(_stop && _pro_buf.empty()) break;
                        //若当前是退出前被换新，或者有数据被唤醒，则返回真，继续向下允许，否则重新陷入休眠
                        _cond_con.wait(lock, [&](){ return _stop || !_pro_buf.empty(); });
                        //退出标志被设置，且生产缓冲区已无数据，这时候再退出，否则有可能会造成生产缓冲区中有数据，但是没有被完全处理
                        _con_buf.swap(_pro_buf);
                        //2.唤醒生产者
                        if(_looper_type == AsyncType::ASYNC_SAFE)
                            _cond_pro.notify_all();
                    }
                    //3.被唤醒后，对消费缓冲区进行数据处理
                    _callBack(_con_buf);
                    //4.初始化消费缓冲区
                    _con_buf.reset();
                }
            }
        private:
            Functor _callBack;//具体 对缓冲区数据进行处理的回调函数，由异步工作器使用者传入
        private:
            AsyncType _looper_type;
            std::atomic<bool> _stop;//工作器停止标志
            Buffer _pro_buf;//生产缓冲区
            Buffer _con_buf;//消费缓冲区
            std::mutex _mutex;
            std::condition_variable _cond_pro;
            std::condition_variable _cond_con;
            std::thread _thread;//异步工作器对应的工作线程
    };
}


#endif